const { Readable, Writable } = require('node:stream');
const fs = require('node:fs');
const path = require('node:path');

// 自定义 Readable 流类
class MyReadable extends Readable {
    constructor(filename, options) {
        super(options);
        this.filename = filename;
        this.fd = null;
        this.buffer = Buffer.alloc(64 * 1024);
        this.position = 0;
        this.open();
    }

    open() {
        console.log('Readable打开文件:', this.filename);
        fs.open(this.filename, 'r', (err, fd) => {
            console.log('fs.open 回调:', err, fd);
            if (err) {
                this.emit('error', err);
                return;
            }
            this.fd = fd;
            this.emit('open'); // 发射 open 事件
            this.read();
        });
    }
    
    _read(size) {
        console.log('_read', size, this.buffer.length);
        // console.log('读取文件:', this.position);
        if (!this.fd) {
            setTimeout(() => {
                this._read(size)
            }, 100)
            return;
        }
        this.buffer = size ? Buffer.alloc(size) : this.buffer;
        fs.read(this.fd, this.buffer, 0, this.buffer.length, this.position, (err, bytesRead) => {
            if (err) {
                this.emit('error', err);
                return;
            }
            if (bytesRead === 0) {
                fs.close(this.fd, (err) => {
                    if (err) this.emit('error', err);
                    this.push(null); // No more data
                });
                return;
            }
            this.position += bytesRead;
            this.push(this.buffer.slice(0, bytesRead));
        });
    }
}

// 自定义 Writable 流类
class MyWritable extends Writable {
    constructor(filename, options) {
        super(options);
        this.filename = filename;
        this.fd = null;
        this.open();
    }

    open() {
        console.log('Writable打开文件:', this.filename);
        fs.open(this.filename, 'w', (err, fd) => {
            console.log('fs.open 回调:', err, fd);
            if (err) {
                this.emit('error', err);
                return;
            }
            this.fd = fd;
            this.emit('open'); // 发射 open 事件
        });
    }

    _write(chunk, encoding, callback) {
        if (!this.fd) {
            this.once('open', () => this._write(chunk, encoding, callback));
            return;
        }
        fs.write(this.fd, chunk, 0, chunk.length, null, (err, bytesWritten) => {
            if (err) {
                this.emit('error', err);
                callback(err);
                return;
            }
            callback();
        });
    }

    _final(callback) {
        if (!this.fd) {
            this.once('open', () => this._final(callback));
            return;
        }
        fs.close(this.fd, (err) => {
            if (err) this.emit('error', err);
            callback(err);
        });
    }
}

// 创建自定义的 Readable 和 Writable 流实例
const myReadable = new MyReadable(path.join(__dirname, './input.txt'));
const myWritable = new MyWritable(path.join(__dirname, './output.txt'));

// 将 Readable 流通过管道传递给 Writable 流
myReadable.pipe(myWritable);

// 监听错误事件
myReadable.on('error', (err) => {
    console.error('读取文件时出错:', err);
});

myWritable.on('error', (err) => {
    console.error('写入文件时出错:', err);
});

myWritable.on('finish', () => {
    console.log('文件传输完成');
});

// 添加日志以调试
myReadable.on('open', () => {
    console.log('Readable 流已打开');
});

myWritable.on('open', () => {
    console.log('Writable 流已打开');
});

myReadable.on('data', (chunk) => {
    console.log(`Received chunk: ${chunk.length} bytes`);
});

myWritable.on('drain', () => {
    console.log('Writable 流已清空');
});
